package pulsar.subs.keyshared;

import org.apache.pulsar.client.api.*;
import pulsar.subs.PulsarProperty;

import java.util.concurrent.TimeUnit;

/**
 * 
 * @version 1.0
 * @date 2021/3/24 17:27
 */
public class MyConsumer {

    static PulsarClient client;

    @SuppressWarnings("all")
    static Consumer consumer;

    static {
        try {
            client = PulsarClient.builder()
                        .serviceUrl(PulsarProperty.serviceUrl)
                        .build();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }


    public static void main(String[] args) throws PulsarClientException {
        subscribe();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                consumer.close();
                client.close();
            } catch (PulsarClientException e) {
                e.printStackTrace();
            }
        }));
    }

    @SuppressWarnings("all")
    static void subscribe() throws PulsarClientException {
        consumer = client.newConsumer()
                .topic(PulsarProperty.topic)
                //订阅组名
                .subscriptionName("keyshared-subscription")
                //指定消费模式，包含：Exclusive，Failover，Shared，Key_Shared。默认Exclusive模式
                .subscriptionType(SubscriptionType.Key_Shared)

                .keySharedPolicy(KeySharedPolicy.stickyHashRange().ranges(Range.of(0,10000)))
                .consumerName("keyshared-consumer-1")
                //指定从哪里开始消费，还有Latest，valueof可选，默认Latest
//                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                //指定消费失败后延迟多久broker重新发送消息给consumer，默认60s
                .negativeAckRedeliveryDelay(60, TimeUnit.SECONDS)
                .subscribe();


        while (true) {
            Message msg = consumer.receive();
            try {
                // Do something with the message
                System.out.println("---------------------------------------------------------------------------");
                System.out.println("Message received: " + new String(msg.getData()));
                System.out.println("---------------------------------------------------------------------------");

                // Acknowledge the message so that it can be deleted by the message broker
                consumer.acknowledge(msg);
            } catch (Exception e) {
                e.printStackTrace();
                // Message failed to process, redeliver later
                consumer.negativeAcknowledge(msg);
            }
        }
    }
}
